package com.baishancloud.log.foglog.sjz

import cn.hutool.json.JSONUtil
import com.baishancloud.log.common.entity.ExternalDataEntity
import com.baishancloud.log.common.env.StreamEnv
import com.baishancloud.log.common.sink.SinkUtil
import com.baishancloud.log.common.source.{BroadcastUtil, SourceUtil}
import com.baishancloud.log.common.util.LogFormatUtil
import com.baishancloud.log.foglog._
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.datastream.BroadcastStream
import org.apache.flink.streaming.api.scala.DataStream
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * 头条单域名数据切换到石家庄 kafka 之后的任务启动类，任务运行在石家庄地区集群
 *
 * @author ziqiang.wang 
 * @date 2022/2/15 10:46
 */
object FoglogDomainStatisSjz extends Serializable {

  def main(args: Array[String]): Unit = {

    val parameterTool: ParameterTool = ParameterTool.fromArgs(args)
    println("输入参数为：" + parameterTool.toMap.toString)
    if (!paramCheck(parameterTool)) {
      return
    }

    StreamEnv.builder(parameterTool).build()

    val foglog1: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, "-foglog-石家庄").uid("cc561109-a8c6-4a6d-bf6b-4f388b69547a")

    // 注意，要记得接受 大S 主题的数据
    val nginx1: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, "-nginx-石家庄").uid("0214e134-aa00-43f3-a1ca-05442d31d1d5")

    val customFetch: ExternalDataEntity = ExternalDataEntity.Builder()
      .enableCountryInfoFetch()
      .enableProvinceInfoFetch()
      .enableIspInfoFetch()
      .enableNodeInfoFetch()
      .enableServiceTypeInfoFetch()
      .enableDomainInfoFetch()
      .enableSvgInfoFetch()
      .build()
    val customInfo: BroadcastStream[ExternalDataEntity] = BroadcastUtil.customInfo(customFetch)

    val foglog: DataStream[(String, String)] = foglog1
      .map(("foglog", _)).name("foglogMap").uid("b634aaec-e350-41ea-8685-cdb110063778")

    val nginx = nginx1
      .map(("nginx", _)).name("nginxMap").uid("c6f29f09-77b3-4fb2-91d8-b505c5229fad")

    nginx.union(foglog)
      .rebalance
      .map(x => (x._1, LogFormatUtil.originParse(x._2))).name("Origin").uid("e3cca883-01b0-4599-a8b2-479f58b0ee90")
      .filter(_._2 != null).name("!=null").uid("841d35a8-2926-43a0-9bfd-0490919f7593")
      .filter(new DomainFilterSjz(parameterTool)).name("domainFilter").uid("5ea355b6-ba63-4fcd-b31b-4d97326184c2")
      .connect(customInfo)
      .process(new LogParse).name("logParse").uid("3e848a1b-99d4-4cdd-939b-3fdd73b331cd")
      .keyBy(_.getKey)
      .window(TumblingProcessingTimeWindows.of(Time.minutes(parameterTool.getLong(windowSize, 1))))
      .reduce((x, y) => Log(
        x.getKey,
        LogValue(
          x.sumRequestNum + y.sumRequestNum,
          x.sumTransBytes + y.sumTransBytes,
          x.sumTransTimeMS + y.sumTransTimeMS,
          x.sumNotRedirectBytes + y.sumNotRedirectBytes,
          x.sumNotRedirectNum + y.sumNotRedirectNum,
          x.sumFirstPacketTimeMS + y.sumFirstPacketTimeMS,
          x.sumFirstFrameNum + y.sumFirstFrameNum,
          x.sumFirstFrameTimeMS + y.sumFirstFrameTimeMS,
          x.sumSslHandshakeTimeMS + y.sumSslHandshakeTimeMS,
          x.sumSslSessionReusedNum + y.sumSslSessionReusedNum,
          x.sumTransSpeedKBps + y.sumTransSpeedKBps,
          x.sumSlow256KBpsNum + y.sumSlow256KBpsNum,
          x.sumSlow512KBpsNum + y.sumSlow512KBpsNum,
          x.sumElapsedWithin1SNum + y.sumElapsedWithin1SNum,
          x.sumElapsedWithin10SNum + y.sumElapsedWithin10SNum
        ))).name("reduce").uid("6d325376-425f-4e8d-a05e-72bbc68a8510")
      .map(JSONUtil.toJsonStr(_)).name("toJson").uid("aed3fc1c-93a8-44ae-9238-e488f72a2378")
      .addSink(SinkUtil.starRocksJsonString(parameterTool)).setParallelism(parameterTool.getInt(sinkParallel, 1)).name("starRocks").uid("5e6a1a0-c262-49ad-8d1e-3e16cce369c4")
    StreamEnv.execute(parameterTool)
  }

}
